package way.learning.zz;

import way.learning.zz.event.AfterStopEvent;
import way.learning.zz.filter.FileFilter;
import way.learning.zz.listener.Rdp4jListener;
import way.learning.zz.source.PolledDirectory;
import org.apache.commons.net.ftp.FTPFile;

import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.concurrent.TimeUnit.MILLISECONDS;


/**
 * 用于添加或者删除 监听器Rdp4jListener和轮询目录 PolledDirectory
 * 并且终止轮询机制
 */
public class DirectoryPoller {
    private static final String NULL_ARGUMENT_ERROR = "Argument is null.";
    private static final long WITH_NO_DELAY = 0;
    private static AtomicInteger threadCount = new AtomicInteger();
    private volatile boolean shouldInvokeShutdownTask = true;

    static final String DEFAULT_THREAD_NAME = "DirectoryPoller-";

    //定时器线程
    private ScheduledRunnable scheduledRunnable;

    //文件过滤器
    private FileFilter filter;
    private long pollingIntervalInMillis;
    private String threadName;

    /**
     * 一个同步计数器,构造时传入int参数,该参数就是计数器的初始值，
     * 每调用一次countDown()方法，计数器减1,计数器大于0 时，await()方法会阻塞程序继续执行
     */
    private CountDownLatch latch = new CountDownLatch(1);

    //线程池
    private ScheduledExecutorService executor;

    //监听器通知者
    ListenerNotifier notifier;
    boolean fileAddedEventEnabledForInitialContent;
    boolean parallelDirectoryPollingEnabled;

    //轮询目录集合
    Set<PolledDirectory> directories;

    Set<FTPFile> fileset;

    //构造函数 DirectoryPollerBuilder是DirectoryPoller的配置类，传入进去
    DirectoryPoller(DirectoryPollerBuilder builder) {
        // First copy values from builder...
        directories = new LinkedHashSet<PolledDirectory>(builder.directories);
        filter = builder.filter;
        pollingIntervalInMillis = builder.pollingIntervalInMillis;
        threadName = builder.threadName;
        fileAddedEventEnabledForInitialContent = builder.fileAddedEventEnabledForInitialContent;
        parallelDirectoryPollingEnabled = builder.parallelDirectoryPollingEnabled;
        notifier = new ListenerNotifier(builder.listeners);

        // ...then check mandatory values
        if (directories.isEmpty()) {
            String pollerName = DirectoryPoller.class.getSimpleName();
            String builderName = DirectoryPollerBuilder.class.getSimpleName();
            String message = "Unable to start the '%s' when No directories has been added! "
                    + "You must add at least one directory before starting the '%s'.\n"
                    + "Call this method to add a directory: %s.addPolledDirectory(PolledDirectory), "
                    + "before you can start the %s.";
            throw new IllegalStateException(String.format(message, pollerName, pollerName, builderName, pollerName));
        }

        if (threadName.equals(DEFAULT_THREAD_NAME)) {
            threadName = threadName + threadCount.incrementAndGet();
        }
        scheduledRunnable = new ScheduledRunnable(this);
    }

    /**
     * 静态方法 构建一个DirectoryPollerBuilder对象
     *
     * 调用 DirectoryPollerBuilder类的无参构造函数
     */
    public static DirectoryPollerBuilder newBuilder() {
        return new DirectoryPollerBuilder();
    }






    /**
     * 启动轮询
     */
    void start() {

        //线程池
        executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName(threadName);
                return t;
            }
        });

        //定期调度任务执行
        executor.scheduleAtFixedRate(scheduledRunnable, WITH_NO_DELAY, pollingIntervalInMillis, MILLISECONDS);
    }









    /**
     * 停止轮询机制。 此方法返回后，将不再有新的轮询周期。
     * 如果在轮询循环期间调用此方法，它将阻塞并等待当前轮询周期完成（这也包括等待任何监听器方法优雅地完成）。
     */
    public void stop() {
        stopAsync();
        awaitTermination();
    }

    /**
     * 停止轮询机制。 此方法返回后，将不再有新的轮询周期。
     * 如果在轮询周期期间调用此方法，它将阻止并等待当前轮询周期完成。 基础线程将被中断（这也包括任何侦听器方法）。
     */
    public void stopNow() {
        stopAsyncNow();
        awaitTermination();
    }

    /**
     * Stops the polling mechanism, but does not wait for any ongoing poll-cycles to finish.
     * Any ongoing poll-cycles will finish gracefully. Use the following method to wait for
     * termination: {@link #awaitTermination()}.
     * <p>
     * Subsequent calls to this method have no affect.
     */
    public void stopAsync() {
        executor.shutdown();
        invokeShutdownTaskOnce();
    }

    /**
     * Stops the polling mechanism, but does not wait for any ongoing poll-cycles to finish.
     * Underlying threads will be interrupted (this also includes any
     * listener methods). Use the following method to wait for termination:
     * {@link #awaitTermination()}.
     * <p>
     * Subsequent calls to this method have no affect.
     */
    public void stopAsyncNow() {
        executor.shutdownNow();
        invokeShutdownTaskOnce();
    }

    /**
     * 调用shutdown任务
     */
    private synchronized void invokeShutdownTaskOnce() {
        if (shouldInvokeShutdownTask) {
            final DirectoryPoller dp = this;

            Util.invokeTask("DP-AfterStop", new Callable<Void>() {
                @Override
                public Void call() {
                    Util.awaitTermination(executor);
                    scheduledRunnable.shutdown();
                    scheduledRunnable.awaitTermination();
                    notifier.afterStop(new AfterStopEvent(dp));
                    latch.countDown();
                    return null;
                }
            });
        }
        shouldInvokeShutdownTask = false;
    }

    /**
     * Blocks until the last poll-cycle has finished and all  AfterStopEvent has been processed.
     */
    public void awaitTermination() {
        try {
            latch.await();
        } catch (InterruptedException e) {
            String message = "awaitTermination() method was interrupted!";
            throw new UnsupportedOperationException(message, e);
        }
    }

    /**
     * @return {@code true} if this {@link DirectoryPoller} is terminated, otherwise {@code false}.
     */
    public boolean isTerminated() {
        return latch.getCount() == 0;
    }

    /**
     * @return the current {@link PolledDirectory}'s handled by
     *         this instance.
     */
    public Set<PolledDirectory> getPolledDirectories() {
        return scheduledRunnable.getDirectories();
    }

    /**
     * @return the polling interval in milliseconds, as
     *         configured for this instance
     */
    public long getPollingIntervalInMillis() {
        return pollingIntervalInMillis;
    }

    /**
     * @return the default {@link FileFilter}, as configured for this
     *         instance.
     */
    public FileFilter getDefaultFileFilter() {
        return filter;
    }

    /**
     * Adds the given <code>listener</code> into this instance.
     * The <code>listener</code> will start receiving notifications
     * in the next coming poll-cycle.
     */
    public void addListener(Rdp4jListener listener) {
        if (listener == null) {
            throw new NullPointerException(NULL_ARGUMENT_ERROR);
        }
        scheduledRunnable.addListener(listener);
    }

    /**
     * Removes the given <code>listener</code> from this instance.
     * The <code>listener</code> will be removed after any ongoing
     * poll-cycles has finished. If there's no ongoing poll-cycles,
     * then the given <code>listener</code> will be removed just before
     * next poll-cycle is started.
     *
     */
    public void removeListener(Rdp4jListener listener) {
        if (listener == null) {
            throw new NullPointerException(NULL_ARGUMENT_ERROR);
        }
        scheduledRunnable.removeListener(listener);
    }

    /**
     * Adds the given <code>directory</code> into this instance.
     * The <code>directory</code> will be polled in the next coming poll-cycle.
     * <p>
     * Registering an already registered directory will be ignored.
     *
     */
    public void addPolledDirectory(PolledDirectory directory) {
        if (directory == null) {
            throw new NullPointerException(NULL_ARGUMENT_ERROR);
        }
        scheduledRunnable.addDirectory(directory);
    }

    /**
     * Removes the given directory from this instance.
     * The directory will be removed after any ongoing poll-cycles has finished.
     * If there's no ongoing poll-cycles,then the given directory will be removed just before next poll-cycle is started.
     *
     */
    public void removePolledDirectory(PolledDirectory directory) {
        if (directory == null) {
            throw new NullPointerException(NULL_ARGUMENT_ERROR);
        }
        scheduledRunnable.removeDirectory(directory);
    }

    /**
     * 以下面这种格式：返回此目录监视器的字符串表示形式
     * "{thread-name}: {directories} [polling every {polling-interval} milliseconds]"
     */
    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder()
                .append(getThreadName())
                .append(": ")
                .append(getPolledDirectories())
                .append(" [polling every: ")
                .append(getPollingIntervalInMillis())
                .append(" milliseconds]");
        return sb.toString();
    }

    /**
     * 返回：相关轮询线程的名称.
     */
    public String getThreadName() {
        return threadName;
    }

    public boolean isParallelDirectoryPollingEnabled() {
        return parallelDirectoryPollingEnabled;
    }

    public boolean isFileAdedEventForInitialContentEnabled() {
        return fileAddedEventEnabledForInitialContent;
    }
}

